package com.flink.demo;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Timestamp;

/**
 * # 本地模式运行
 * java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.UserBehaviorAnalysis
 * <p>
 * # 带参数运行示例
 * java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.UserBehaviorAnalysis --host 192.40.10.112 --port 9999
 * <p>
 * Flink 数据统计程序
 * 功能: 1. 实时用户行为统计 2. 热门商品统计 3. 分类购买量统计
 */
public class UserBehaviorAnalysis {
    private static final Logger logger = LoggerFactory.getLogger(UserBehaviorAnalysis.class);

    public static void main(String[] args) throws Exception {
        logger.info("应用程序启动");

        // 解析命令行参数
        String host = "192.40.10.112";
        int port = 9999;

        for (int i = 0; i < args.length; i++) {
            switch (args[i]) {
                case "--host":
                    if (i + 1 < args.length) host = args[++i];
                    break;
                case "--port":
                    if (i + 1 < args.length) {
                        try {
                            port = Integer.parseInt(args[++i]);
                        } catch (NumberFormatException e) {
                            logger.error("端口号格式错误，使用默认端口 9999");
                            port = 9999;
                        }
                    }
                    break;
                case "--help":
                    printUsage();
                    return;
            }
        }


        logger.info("配置 - Host: " + host + ", Port: " + port);

        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 如果不设置，Flink会使用WebUI中配置的并行度
        logger.info("当前并行度配置: " + env.getParallelism());

        // 固定延迟重启策略（尝试重启3次，每次间隔10秒）
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, // 重启次数
                org.apache.flink.api.common.time.Time.seconds(10) // 间隔时间
        ));

        // 或故障率重启策略
        env.setRestartStrategy(RestartStrategies.failureRateRestart(
                3, // 每个时间间隔允许的最大失败次数
                org.apache.flink.api.common.time.Time.minutes(5), // 时间间隔
                org.apache.flink.api.common.time.Time.seconds(10) // 重启间隔
        ));

        // 从Socket读取数据（使用参数配置）
        DataStream<String> textStream = env.socketTextStream(host, port, "\n");

        // 解析数据流
        DataStream<UserBehavior> dataStream = textStream.flatMap(new FlatMapFunction<String, UserBehavior>() {
            @Override
            public void flatMap(String value, Collector<UserBehavior> out) throws Exception {
                try {
                    // 简化解析，实际应该更严谨
                    if (value.contains("UserBehavior")) {
                        String[] parts = value.split(",");
                        if (parts.length >= 5) {
                            Long userId = Long.parseLong(parts[0].split("=")[1]);
                            String itemId = parts[1].split("='")[1].replace("'", "");
                            String category = parts[2].split("='")[1].replace("'", "");
                            String behavior = parts[3].split("='")[1].replace("'", "");
                            Timestamp timestamp = new Timestamp(System.currentTimeMillis());

                            out.collect(new UserBehavior(userId, itemId, category, behavior, timestamp));
                        }
                    }
                } catch (Exception e) {
                    logger.error("解析数据失败: " + value);
                }
            }
        });

        // 1. 统计每种行为的数量（每10秒窗口）
        DataStream<Tuple2<String, Integer>> behaviorCounts = dataStream
                .map(new MapFunction<UserBehavior, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(UserBehavior behavior) throws Exception {
                        return new Tuple2<>(behavior.getBehavior(), 1);
                    }
                })
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .sum(1)/*.setParallelism(2)*/;//设置子任务并行度

        // 2. 统计热门商品（点击量前5）
        DataStream<Tuple2<String, Integer>> popularItems = dataStream
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior behavior) throws Exception {
                        return "click".equals(behavior.getBehavior());
                    }
                })
                .map(new MapFunction<UserBehavior, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(UserBehavior behavior) throws Exception {
                        return new Tuple2<>(behavior.getItemId(), 1);
                    }
                })
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
                .sum(1);

        // 3. 统计每个分类的购买量
        DataStream<Tuple2<String, Integer>> categorySales = dataStream
                .filter(new FilterFunction<UserBehavior>() {
                    @Override
                    public boolean filter(UserBehavior behavior) throws Exception {
                        return "buy".equals(behavior.getBehavior());
                    }
                })
                .map(new MapFunction<UserBehavior, Tuple2<String, Integer>>() {
                    @Override
                    public Tuple2<String, Integer> map(UserBehavior behavior) throws Exception {
                        return new Tuple2<>(behavior.getCategory(), 1);
                    }
                })
                .keyBy(0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(10)))
                .sum(1);

        // 打印结果  前面的数字是并行度
        behaviorCounts.print("行为统计");//行为统计:4> (pv,1)
        popularItems.print("热门商品");//热门商品:13> (item_306,1)
        categorySales.print("分类销量");//分类销量:3> (sports,1)

        // 执行任务
        env.execute("User Behavior Analysis");
    }

    /**
     * 打印使用说明
     */
    private static void printUsage() {
        logger.info("Usage: java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.UserBehaviorAnalysis [options]");
        logger.info("Options:");
        logger.info("  --host <hostname>    Socket主机地址 (默认: localhost)");
        logger.info("  --port <port>        Socket端口号 (默认: 9999)");
        logger.info("  --parallelism <n>    全局并行度");
        logger.info("  --help              显示帮助信息");
        logger.info("\nExamples:");
        logger.info("  java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.UserBehaviorAnalysis");
        logger.info("  java -cp target/flink-demo-1.0-SNAPSHOT-fat.jar com.flink.demo.UserBehaviorAnalysis --host 192.168.1.100 --port 8888");
    }
}